package com.vpen.kafkademo.kafka.consumer;

import com.vpen.kafkademo.model.entity.StoreContentDownload;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 消费者
 *
 * @author by Vpen
 * @date 2021/8/24 19:24
 */
@Component
@Slf4j
public class KafkaConsumer {

    private final String TOPIC = "testTopic";


    @KafkaListener(topics = {"testTopic"}, groupId = "Consumer" + "-testTopic")
    public void consumer(ConsumerRecord<?, ?> record) {
        Optional<?> msg = Optional.ofNullable(record.value());
        msg.ifPresent(o -> log.info("接收到消息：{}", o));
        msg.ifPresent(o -> log.info("topic：{}", record.topic()));
        msg.ifPresent(o -> log.info("value：{}", record.value()));
        msg.ifPresent(o -> log.info("partition：{}", record.partition()));
    }

    /***
     * 描述:  @KafkaListener 注解 指定了消费的 Topic 和 消费者分组 。
     * @author Vpen
     * @date 2021/8/25 11:12
     *
     * @param scd
     * @return void
     **/
    @KafkaListener(topics = {"testTopic"}, groupId = "Consumer" + "-testTopic")
    public void consumer1(StoreContentDownload scd) {
        log.info("线程{}--download:{}", Thread.currentThread().getName(), scd);
    }


    @KafkaListener(topics = "VpenTopic", groupId = "C" + "-testTopic")
    public void consumer2(StoreContentDownload scd, ConsumerRecord<?, StoreContentDownload> record) {
        log.info("线程{}--download:{}", Thread.currentThread().getName(), scd);

        StoreContentDownload storeContentDownload = record.value();
        // 主题
        String topic = record.topic();
        // 分区
        int partition = record.partition();

        Object key = record.key();
        long offset = record.offset();

        log.info("offset:{}",offset);
        log.info("key:{}",key);
        log.info("topic：{}", topic);
        log.info("value：{}", storeContentDownload);
        log.info("partition：{}", partition);
    }

}
